package com.dz.time;

import com.dz.utils.ConcurrentDateUtil;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import time.pproduct.Products;

/**
 * @ClassName AggregateFunctionMain2
 * @Description TODO
 * @Author zhangdong
 * @Date 2020/11/29 18:46
 * @Version 1.0
 */
public class AggregateFunctionMain2
{


    public  static int windowSize=6000;/*滑动窗口大小*/

    public  static int windowSlider=3000;/*滑动窗口滑动间隔*/

    public  static  String sourcePath ="D:\\git_code\\hatchflink\\flink-advance\\src\\main\\resources\\products\\tt.txt";

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);
        //使用事件时间
        senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<String> sourceData = senv.readTextFile(sourcePath);
        DataStream<Products> productsDataStream = sourceData.map(new MapFunction<String, Products>() {
            @Override
            public Products map(String value) throws Exception {
                Object[] conents = value.split(",");

                return new Products(conents[0].toString(),conents[1].toString(),Long.valueOf(conents[2].toString()),
                        Long.valueOf(conents[3].toString()));
            }
        }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Products>() {
            //时间戳
            @Override
            public long extractAscendingTimestamp(Products element) {
                return element.getTime();
            }
        });

        KeyedStream<Products,String> productViewCount = productsDataStream.filter(new FilterFunction<Products>() {
            @Override
            public boolean filter(Products value) throws Exception {
                if(value.getViewCount()>0)
                {
                   return true;
                }
                return  false;
            }
        }).keyBy(new KeySelector<Products, String>() {
            @Override
            public String getKey(Products value) throws Exception {
                return value.getProductName();
            }
        });

        productViewCount.timeWindow(Time.milliseconds(windowSize), Time.milliseconds(windowSlider))
        .aggregate(new AggregateFunction<Products, Long, Long>() {
            @Override
            public Long createAccumulator() {
                return 0l;
            }

            @Override
            public Long add(Products value, Long accumulator) {
                return accumulator+1;
            }

            @Override
            public Long getResult(Long accumulator) {
                return accumulator;
            }

            @Override
            public Long merge(Long a, Long b) {
                return a+b;
            }
        }, new WindowFunction<Long , String, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<Long> input, Collector<String> out) throws Exception {

                out.collect("----------------窗口时间："+ window.getEnd()+"==="+ ConcurrentDateUtil
                        .dateFormat(window.getEnd()));
                out.collect("商品ID: "+s+"  浏览量: "+input.iterator().next());
            }
        }).print();

        senv.execute("eee");
    }
}
